package io.kiki.sba.registry.server.metadata.slot.tasks;


import io.kiki.sba.registry.common.model.metaserver.nodes.DataStoreNode;
import io.kiki.sba.registry.common.model.slot.Slot;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.server.metadata.slot.ReBalanceTask;
import io.kiki.sba.registry.server.metadata.slot.SlotManager;
import io.kiki.sba.registry.util.DatumVersionUtil;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;


public class ReBalanceTaskImpl implements ReBalanceTask {

    private static final Logger logger = LoggerFactory.getLogger(ReBalanceTaskImpl.class);

    private final SlotManager slotManager;

    private final AtomicInteger nextLeaderIndex = new AtomicInteger();

    private final AtomicInteger nextFollowerIndex = new AtomicInteger(1);

    private final List<DataStoreNode> dataStoreNodeList;

    public ReBalanceTaskImpl(SlotManager slotManager, List<DataStoreNode> dataStoreNodeList) {
        this.slotManager = slotManager;
        this.dataStoreNodeList = new ArrayList<>(dataStoreNodeList);
    }

    @Override
    public void run() {
        if (CollectionUtils.isEmpty(dataStoreNodeList)) {
            logger.info("[run] empty candidate, quit");
            return;
        }
        if (logger.isInfoEnabled()) {
            logger.info("[run] candidates({}): {}", dataStoreNodeList.size(), dataStoreNodeList);
        }
        initParameters();
        Slots slots = createSlots();
        if (logger.isInfoEnabled()) {
            logger.info("[run] end to init slot table");
        }

        slotManager.refresh(slots);
        if (logger.isInfoEnabled()) {
            logger.info("[run] raft refreshed slot-table");
        }
    }

    private void initParameters() {
        nextLeaderIndex.set(0);
        nextFollowerIndex.set(dataStoreNodeList.size() - 1);
    }

    private Slots createSlots() {
        Map<Integer, Slot> slotMap = generateSlotMap();
        return new Slots(DatumVersionUtil.nextId(), slotMap.values());
    }

    private Map<Integer, Slot> generateSlotMap() {
        Map<Integer, Slot> slotMap = new HashMap<>();
        for (int i = 0; i < slotManager.getSlotNums(); i++) {
            long epoch = DatumVersionUtil.nextId();
            String leader = getNextLeader().getIp();
            List<String> followers = new ArrayList<>();
            for (int j = 0; j < slotManager.getSlotReplicaNums() - 1; j++) {
                followers.add(getNextFollower().getIp());
            }
            Slot slot = new Slot(i, leader, epoch, followers);
            slotMap.put(i, slot);
        }
        return slotMap;
    }

    private DataStoreNode getNextLeader() {
        return dataStoreNodeList.get(nextLeaderIndex.getAndIncrement() % dataStoreNodeList.size());
    }

    private DataStoreNode getNextFollower() {
        return dataStoreNodeList.get(nextFollowerIndex.getAndIncrement() % dataStoreNodeList.size());
    }
}
